package raft

// The file raftapi/raft.go defines the interface that raft must
// expose to servers (or the tester), but see comments below for each
// of these functions for more details.
//
// Make() creates a new raft peer that implements the raft interface.

import (
//	"bytes"
// "math/rand"
// "sync"
// "sync/atomic"
// "time"

//	"6.5840/labgob"
//
// "6.5840/labrpc"
// "6.5840/raftapi"
// "6.5840/tester1"
)

type requestVoteMsg struct {
	rpcServiceBaseMsg
	args       *RequestVoteArgs
	resChannel chan<- int
}

// example RequestVote RPC arguments structure.
// field names must start with capital letters!
type RequestVoteArgs struct {
	// Your data here (3A, 3B).
	Term         int
	CandidateId  int
	LastLogIndex int
	LastLogTerm  int
}

// example RequestVote RPC reply structure.
// field names must start with capital letters!
type RequestVoteReply struct {
	// Your data here (3A).
	Term        int
	VoteGranted bool
}

// 1: 1新，0：一样新;-1：2新
func compareUpToDate(logTerm1, logIndex1, logTerm2, logIndex2 int) int {
	if logTerm1 > logTerm2 {
		return 1
	}
	if logTerm2 > logTerm1 {
		return -1
	}
	if logIndex1 > logIndex2 {
		return 1
	}
	if logIndex2 > logIndex1 {
		return -1
	}
	return 0
}

// example RequestVote RPC handler.
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (3A, 3B).
	needPersist := false
	rf.mu.Lock()
	defer rf.mu.Unlock()
	if args.Term < rf.CurrentTerm {
		reply.Term = rf.CurrentTerm
		reply.VoteGranted = false
		return
	}

	if args.Term > rf.CurrentTerm {
		rf.CurrentTerm = args.Term
		rf.becomeFollower()
		needPersist = true
	}

	lastLogIndex := rf.relative2Abs(len(rf.Log) - 1)
	lastLogTerm := rf.Log[len(rf.Log)-1].Term
	if (rf.VotedFor == VOTED_NOBODY || rf.VotedFor == args.CandidateId) && compareUpToDate(args.LastLogTerm, args.LastLogIndex, lastLogTerm, lastLogIndex) >= 0 {
		rf.VotedFor = args.CandidateId
		rf.electionTimeout = false
		reply.Term = args.Term
		reply.VoteGranted = true
		needPersist = true
		// Debug(dVote, "S%d Granting Vote to S%d at T%d", rf.me, args.CandidateId, rf.CurrentTerm)
	}
	if needPersist {
		rf.persist()
	}
}

// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

// 向特定的peer拉票，如果获得了票，resChannel发送这个peer所投票的term
func (rf *Raft) requestVoteFrom(server int, args *RequestVoteArgs, resChannel chan<- int) {
	reply := &RequestVoteReply{}
	// reply的term保证至少和args里的一样大
	ok := rf.sendRequestVote(server, args, reply)
	if !ok {
		return
	}
	rf.mu.Lock()
	defer rf.mu.Unlock()
	if reply.Term > rf.CurrentTerm {
		rf.becomeFollower()
		return
	}
	// 这里是reply.Term <= rf.CurrentTerm的情况，electionService里会统一处理的
	if reply.VoteGranted {
		go func() { resChannel <- reply.Term }()
	}
}

func (rf *Raft) startRequestVote(msg *requestVoteMsg) {
	for i := range msg.n_peers {
		if i != msg.me {
			go rf.requestVoteFrom(i, msg.args, msg.resChannel)
		}
	}
}

// 只提供election服务，不提供身份转变。身份转变由RPC和tricker提供
func (rf *Raft) electionService() {
	voteCount := 0
	resChannel := make(chan int)
	rf.mu.Lock()
	n_peers := len(rf.peers)
	rf.mu.Unlock()

	for !rf.killed() {
		select {
		// 发起election
		case electionStartTerm := <-rf.startElectionChannel:
			voteCount = 0
			rf.mu.Lock()
			if rf.status == CANDIDATE && electionStartTerm == rf.CurrentTerm {
				rf.CurrentTerm++
				rf.VotedFor = rf.me
				voteCount = 1
				rf.electionTimeout = false
				args := &RequestVoteArgs{Term: rf.CurrentTerm, CandidateId: rf.me, LastLogIndex: rf.relative2Abs(len(rf.Log) - 1), LastLogTerm: rf.Log[len(rf.Log)-1].Term}
				// request vote from
				msg := &requestVoteMsg{rpcServiceBaseMsg: rpcServiceBaseMsg{rf.me, n_peers}, args: args, resChannel: resChannel}

				go rf.startRequestVote(msg)
				// Debug(dVote, "S%d requesting votes from peers", rf.me)
				rf.persist()
			}
			rf.mu.Unlock()
		// 投票
		case electionTerm := <-resChannel:
			rf.mu.Lock()
			// Debug(dVote, "S%d receive a vote in T%d, expect T%d", rf.me, rf.CurrentTerm, electionTerm)
			if rf.status != CANDIDATE || electionTerm != rf.CurrentTerm {
				rf.mu.Unlock()
				continue
			}
			voteCount++
			// Debug(dVote, "S%d Get a vote. Now is %d votes.", rf.me, voteCount)
			if (voteCount << 1) > n_peers {
				// Debug(dLeader, "S%d Achieved Majority for T%d (%d), converting to Leader", rf.me, rf.CurrentTerm, voteCount)
				go rf.becomeLeader(rf.CurrentTerm)

			}
			rf.mu.Unlock()
		}
	}
	// Debug(dLog, "S%d is killed. electionService() is stopped.", rf.me)
}

func (rf *Raft) startElectionService(sign int) {
	rf.startElectionChannel <- sign
}

// === status convertion ===

// 这个里面有加锁
// 只提供status转变和相关初始化，不改变Term
func (rf *Raft) becomeLeader(term int) {
	// term := <-rf.startInitLeaderChannel
	rf.mu.Lock()
	defer rf.mu.Unlock()
	// Debug(dLeader, "S%d initing Leader, status %d", rf.me, rf.status)
	if rf.status != CANDIDATE || rf.CurrentTerm != term {
		return
	}
	rf.status = LEADER
	// Debug(dLeader, "S%d become Leader of T%d. State is %v", rf.me, rf.CurrentTerm, rf)
	// Debug(dLeader, "S%d become Leader of T%d, abslen(Log) %d", rf.me, rf.CurrentTerm, rf.relative2Abs(len(rf.Log)))
	// Debug(dLeader, "S%d try to start heartbeat", rf.me)
	go rf.startHeartbeatService(rf.CurrentTerm)
	absLogLength := rf.relative2Abs(len(rf.Log))
	for i := range rf.nextIndex {
		// Debug(dLog, "S%d ,T%d, set S%d, init nextIndex %d", rf.me, rf.CurrentTerm, i, len(rf.Log))
		rf.nextIndex[i] = absLogLength
	}
	for i := range rf.matchIndex {
		rf.matchIndex[i] = 0
	}
}

// 这个要加锁后用
// 只提供status转变和相关初始化，不改变Term
func (rf *Raft) becomeCandidate() {
	rf.status = CANDIDATE
	// Debug(dInfo, "S%d become Candidate in T%d, starting election. State is %v", rf.me, rf.CurrentTerm, rf)
	go rf.startElectionService(rf.CurrentTerm)
}

// 这个要加锁后用
// 只提供status转变和相关初始化，不改变Term
func (rf *Raft) becomeFollower() {
	rf.status = FOLLOWER
	rf.VotedFor = VOTED_NOBODY
	rf.lastConsistentIndex = 0
	// Debug(dLeader, "S%d become Follower of T%d, abslen(Log) %d", rf.me, rf.CurrentTerm, rf.relative2Abs(len(rf.Log)))
}
